package com.shujia.flink.window;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author shujia
 */
public class Demo9windowProcess {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> lines = env.socketTextStream("master", 8888);

        DataStream<Tuple2<String, Integer>> kv = lines.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyBy = kv.keyBy(k -> k.f0);

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windows = keyBy
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));


        /*
         * process算子是flink一个底层的算子,用于做很复杂逻辑计算
         */

        SingleOutputStreamOperator<CountResult> countResults = windows
                .process(new ProcessWindowFunction<Tuple2<String, Integer>, CountResult, String, TimeWindow>() {
                    /**
                     * process: 每一个key每个窗口执行一次
                     * @param word 分组的key
                     * @param context 上下文对象,可以获取flink的运行环境配置,可以获取窗口的开始和结束时间
                     * @param elements :窗口内所有的数据
                     * @param out 用于将数据发送到下游
                     */
                    @Override
                    public void process(String word,
                                        ProcessWindowFunction<Tuple2<String, Integer>, CountResult, String, TimeWindow>.Context context,
                                        Iterable<Tuple2<String, Integer>> elements,
                                        Collector<CountResult> out)  {
                        //获取窗口的开始和结束时间
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        //获取当前处理时间
                        long processingTime = context.currentProcessingTime();
                        //获取当前水位线
                        long watermark = context.currentWatermark();

                        int count = 0;
                        //计算单词的数量
                        for (Tuple2<String, Integer> element : elements) {
                            count++;
                        }
                        //将数据发送到下游
                        out.collect(new CountResult(word, count, start, end, processingTime, watermark));
                    }
                });

        countResults.print();

        env.execute();

    }

    @Data
    @AllArgsConstructor
    static class CountResult {
        String word;
        int count;
        long start;
        long end;
        long processingTime;
        long watermark;
    }
}
